package com.gis.aoitest.window;

import com.gis.aoitest.beans.SensorReading;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;


public class TumblingEventWindows2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        // socket文本流
        DataStreamSource<String> inputStream = env.socketTextStream("localhost", 9999);

        // 转换成SensorReading类型
        SingleOutputStreamOperator<Tuple3<String, Long, Double>> dataStream = inputStream.map(line -> {
                    String[] fields = line.split(",");
                    return new Tuple3<String, Long, Double>(fields[0], Long.parseLong(fields[1]), Double.parseDouble(fields[2]));
//                    return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
                })
                .returns(Types.TUPLE(Types.STRING, Types.LONG, Types.DOUBLE))
                // 时间语义
                // 升序数据设置时间戳和watermark
//            .assignTimestampsAndWatermarks(new AscendingTimestampExtractor<SensorReading>() {
//                @Override
//                public long extractAscendingTimestamp(SensorReading element) {
//                    return element.getTimestamp() * 1000L;
//                }
//            })
                // 乱序数据设置时间戳和watermark
                .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Tuple3<String, Long, Double>>(Time.seconds(1)) {
                    @Override
                    public long extractTimestamp(Tuple3<String, Long, Double> element) {
                        return element.f1 * 1000L;
                    }
                });

        // OutputTag<SensorReading> outputTag = new OutputTag<SensorReading>("late"){};

        // 基于时间的开窗集合，统计15s内，温度的最小值
        SingleOutputStreamOperator<Tuple3<String, Long, Double>> minTempStream = dataStream.keyBy(0)
                //.timeWindow(Time.seconds(15))
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
//                .allowedLateness(Time.minutes(1))
//                .sideOutputLateData(outputTag)
                .minBy(2);

        minTempStream.print("minTemp");
        // minTempStream.getSideOutput(outputTag).print();

        env.execute();
    }
}
